package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class DwsTrafficVcChArlsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 2.读取三个主题的数据创建流
        String uvTopic="dwd_traffic_unique_visitor_detail";
        String ujdTopic="dwd_traffic_user_jump_detail";
        String topic="dwd_traffic_page_log";
        String groupId="vccharisnew_pageview_window_2103a";

        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> ujDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(ujdTopic, groupId));
        DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo 3.统一数据格式
        //独立
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUvDS
                = uvDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            return new TrafficPageViewBean(
                    null,
                    null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L, 0L, 0L, 0L, 0L,
                    jsonObject.getLong("ts"));
        });


        //跳出
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUjDS = ujDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 0L, 0L, 0L, 1L,
                    jsonObject.getLong("ts"));
        });

        //会话数 last_page_id=null 1.页面浏览数 1 和 浏览总时长 during_time
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithPageDS = pageDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            JSONObject page = jsonObject.getJSONObject("page");
            String lastPageId = page.getString("last_page_id");
            long sv = 0L; //会话数
            if (lastPageId == null) {
                sv = 1L;
            }
            return new TrafficPageViewBean(null, null,
                    common.getString("vc"), //版本
                    common.getString("ch"), //驱动
                    common.getString("ar"), //区域
                    common.getString("is_new"), //新老用户
                    0L, //独立
                    sv,//会话
                    1L,
                    page.getLong("during_time"),
                    0L,
                    jsonObject.getLong("ts"));
        });

        trafficPageViewWithUjDS.print(">>>>>>>>>>>>ujCTujCTujCT>>>>>>>>>>>>>>");
        trafficPageViewWithUvDS.print("独立会话数");
        trafficPageViewWithPageDS.print("--page-->");

        //todo 4.将三个流进行Union
        DataStream<TrafficPageViewBean> unionDS = trafficPageViewWithUvDS.union(
                trafficPageViewWithUjDS,
                trafficPageViewWithPageDS);

        //todo 5.提取事件时间生成WaterMark
        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithWmDS = unionDS
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                        return trafficPageViewBean.getTs();
                    }
                }));
        trafficPageViewWithWmDS.print("查看水位线------------->>>>");

        //todo 6.分组开窗聚合
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowedStream = trafficPageViewWithWmDS
                .keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {

                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean v) throws Exception {
                        return Tuple4.of(v.getAr(),v.getCh(),v.getIsNew(),v.getVc());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(10l));


        SingleOutputStreamOperator<TrafficPageViewBean> resultDS = windowedStream
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean v1, TrafficPageViewBean v2) throws Exception {
                v1.setSvCt(v1.getSvCt() + v2.getSvCt());
                v1.setUvCt(v1.getUvCt() + v2.getUvCt());
                v1.setUjCt(v1.getUjCt() + v2.getUjCt());
                v1.setPvCt(v1.getPvCt() + v2.getPvCt());
                v1.setDurSum(v1.getDurSum() + v2.getDurSum());
                return v1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple4<String, String, String, String> key, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                //获取数据
                TrafficPageViewBean next = input.iterator().next();
                //补充信息
                next.setStt(new Timestamp(window.getStart()));
                next.setEdt(new Timestamp(window.getEnd()));
                //修改TS
                next.setTs(System.currentTimeMillis());
                //输出数据
                out.collect(next);
            }
        });

        //todo 7.将数据写出到ClickHouse
        resultDS.print(">>>>>>>>>>>>");
        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        //todo 8.启动任务
        env.execute("DwsTrafficVcChArIsNewPageViewWindow");


    }
}
